package net.xuele.learn.flink.example;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;

import java.util.Collections;
import java.util.List;

/**
 * @Author patrick
 * @Date 2023/7/12 18:32
 * @Description: ListCheckpointed的使用，可以看到算子的状态被保留了
 */
public class ListCheckpointedTest<T> implements MapFunction<T, Tuple2<T, Long>>, ListCheckpointed<Long> {

    private long count;

    @Override
    public Tuple2<T, Long> map(T value) throws Exception {
        count++;
        return new Tuple2<>(value, count);
    }

    /**
     * 通过阅读源代码，发现返回值会被存储起来
     * @param checkpointId
     * @param timestamp
     * @return
     * @throws Exception
     */
    @Override
    public List<Long> snapshotState(long checkpointId, long timestamp) throws Exception {
        return Collections.singletonList(count);
    }

    /**
     * 会尝试从状态后端获取状态数据，再调用此接口
     * @param state
     * @throws Exception
     */
    @Override
    public void restoreState(List<Long> state) throws Exception {
        for (Long l : state) {
            count += l;
        }
    }
}
